package main

import (
	"fmt"
	"sync"
)

//// 流水线进阶: 扇入和扇出
//扇出：同一个 channel 可以被多个函数读取数据，直到channel关闭。
//这种机制允许将工作负载分发到一组worker，以便更好地并行使用 CPU 和 I/O。
//
//扇入：多个 channel 的数据可以被同一个函数读取和处理，然后合并到一个 channel，直到所有 channel都关闭。

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

func sq(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

func merge(cs ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// 为每一个输入 channel cs 创建一个 goroutine output
	// output 将数据从 c 拷贝到 out,直到 c 关闭,然后调用 wg.Done
	output := func(c <-chan int) {
		for n := range c {
			out <- n
		}
		wg.Done()
	}
	wg.Add(len(cs))
	for _, c := range cs {
		go output(c)
	}

	// 启动一个 goroutine，用于所有 output goroutine结束时，关闭 out
	// 该goroutine 必须在 wg.Add 之后启动
	go func() {
		wg.Wait()
		close(out)
	}()

	return out

}

func main() {
	in := gen(2, 3)
	// 启动两个 sql 实例,即两个 goroutines 处理 channel "in" 的数据
	c1 := sq(in)
	c2 := sq(in)

	// merge 函数将 channel c1和 c2 合并到一起,这段代码会消费 merge 的结果
	for n := range merge(c1, c2) {
		fmt.Println(n) // // 打印 4 9, 或 9 4
	}
}
